package com.ybliang.plat.common.rpc.consumer.netty;

import com.ybliang.plat.common.rpc.core.DefaultFuture;
import com.ybliang.plat.common.rpc.core.RpcRequest;
import com.ybliang.plat.common.rpc.core.RpcResponse;
import com.ybliang.plat.common.rpc.protocol.json.JSON;
import io.netty.channel.ChannelDuplexHandler;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelPromise;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.net.InetSocketAddress;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

@ChannelHandler.Sharable
public class NettyClientHandler extends ChannelDuplexHandler {

    public static Logger LOGGER = LoggerFactory.getLogger(NettyClientHandler.class);

    /**
     * 使用Map维护请求对象ID与响应结果Future的映射关系
     */
    private final Map<String, DefaultFuture> futureMap = new ConcurrentHashMap<>();

    @Override
    public void channelActive(ChannelHandlerContext ctx)   {
        LOGGER.info("NettyClientHandler->channelActive 已连接到RPC服务器.{}",ctx.channel().remoteAddress());
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx)   {
        InetSocketAddress address =(InetSocketAddress) ctx.channel().remoteAddress();
        LOGGER.info("NettyClientHandler->channelInactive 与RPC服务器断开连接:{},准备执行重连操作", address);
        ctx.channel().close();
    }

    @Override
    public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
        LOGGER.info("NettyClientHandler->write 准备处理 msg:{}...", JSON.toJSONString(msg));
        if (msg instanceof RpcRequest) {
            RpcRequest request = (RpcRequest) msg;
            LOGGER.info("NettyClientHandler->write 把requestId:{}写入futureMap", request.getId());
            //发送请求对象之前，先把请求ID保存下来，并构建一个与响应Future的映射关系
            futureMap.putIfAbsent(request.getId(), new DefaultFuture());
        }
        super.write(ctx, msg, promise);
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        LOGGER.info("NettyClientHandler->channelRead 收到数据 msg:{}...", JSON.toJSONString(msg));
        if(msg instanceof RpcResponse){
            RpcResponse response = (RpcResponse) msg;
            DefaultFuture future = futureMap.get(response.getId());
            if(null == future){
                future = new DefaultFuture();
            }
            future.setResponse(response);
            futureMap.put(response.getId(), future);
        }
        super.channelRead(ctx, msg);
    }



    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        LOGGER.info("NettyClientHandler -> userEventTriggered 已超过30秒未与RPC服务器进行读写操作");
        if (evt instanceof IdleStateEvent){
            IdleStateEvent event = (IdleStateEvent)evt;
            if (event.state()== IdleState.ALL_IDLE){
                LOGGER.info("NettyClientHandler -> userEventTriggered 将发送心跳消息...");
                RpcRequest request = new RpcRequest();
                request.setMethodName("heartBeat");
                ctx.channel().writeAndFlush(request);
            }
        }else{
            super.userEventTriggered(ctx, evt);
        }
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause){
        LOGGER.info("NettyClientHandler -> exceptionCaught RPC通信服务器发生异常.{}", cause);
        ctx.channel().close();
    }

    /**
     * 获取RpcResponse
     * @param requestId
     * @return
     */
    public RpcResponse getRpcResponse(String requestId) {
        try {
            DefaultFuture future = futureMap.get(requestId);
            if(null == future){
                return null;
            }
            return future.getRpcResponse(5000);
        } finally {
            //获取成功以后，从map中移除
            futureMap.remove(requestId);
        }
    }
}
